package bold.deposit.useful;

import org.junit.Test;
import redis.clients.jedis.*;
import redis.clients.util.JedisClusterCRC16;

import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 由于 redis集群的solr 需要计算到对应节点
 * 增加redis 集群，批量写入
 * 只能单线程运行解析数据，如果需要多线程解析数据 -> 可以现将数据 queue 聚合操作 然后 while循环 操作
 *
 * @author khaiLoon
 * @date 2018/9/14
 **/
public class JedisPipelineUtil {

    public static void main(String[] args) {
        String ipHost =
                "127.0.0.1:7001," +
                "127.0.0.1:7002," +
                "127.0.0.1:7003";

        RedisBind<String> redisBind = new RedisBind<>(ipHost);

        for (String s : getData()) {
            RedisBind<String>.PollInfo pollInfo = redisBind.addData(s);
            List<String> list = pollInfo.getDataList();
            if (list.size() >= 200) {
                Jedis jedis = pollInfo.getJedisPool().getResource();
                Pipeline p = jedis.pipelined();

                for (String key : list) {
                    p.set(key, "value");
                }

                p.sync();
                jedis.close();
                list.clear();
            }
        }

        for (RedisBind<String>.PollInfo pollInfo : redisBind.getFlush()) {
            List<String> data = pollInfo.getDataList();
            Jedis jedis = pollInfo.getJedisPool().getResource();
            Pipeline p = jedis.pipelined();
            for (String key : data) {
                p.set(key, "value");
            }
            p.sync();
            jedis.close();
        }

        redisBind.close();
        System.out.println("批量保存数据=" + redisBind.getCount());
    }

    /**
     * 随机key
     */
    private static List<String> getData() {
        ArrayList<String> list = new ArrayList<>(1000);
        for (int i = 0; i < 2000; i++) {
            list.add(String.format("LZ%s", i));
        }
        return list;
    }


    static class RedisBind<T> {
        private AtomicLong count = new AtomicLong();
        private TreeMap<Long, PollInfo> slotTree = new TreeMap<>();


        public RedisBind(String ipHost) {
            JedisCluster jedisCluster = getJedisCluster(ipHost);
            initSlotTree(jedisCluster);
        }

        public PollInfo addData(T data) {
            int slot = JedisClusterCRC16.getSlot(data.toString());
            Map.Entry<Long, PollInfo> entry = slotTree.floorEntry((long) slot);//临界值处理没问题
            PollInfo pollInfo = entry.getValue();
            pollInfo.getDataList().add(data);
            count.incrementAndGet();
            return pollInfo;
        }

        public Collection<PollInfo> getFlush() {
            return slotTree.values();
        }

        private JedisCluster getJedisCluster(String ipHost) {
            String[] ipArr = ipHost.split(",");
            Set<HostAndPort> nodes = new HashSet<>();
            for (String hostAndPost : ipArr) {
                String[] split = hostAndPost.split(":");
                HostAndPort hostAndPort = new HostAndPort(split[0], Integer.parseInt(split[1]));
                nodes.add(hostAndPort);
            }
            return new JedisCluster(nodes);
        }

        public long getCount() {
            return count.get();
        }

        public void close() {
            for (PollInfo pollInfo : slotTree.values()) {
                pollInfo.getJedisPool().close();
            }
        }

        private void initSlotTree(JedisCluster jedisCluster) {
            Map<String, JedisPool> nodePollMap = jedisCluster.getClusterNodes();

            Jedis jedis = nodePollMap.values().iterator().next().getResource();
            List<Object> list = jedis.clusterSlots();
            for (Object object : list) {
                List list1 = (List) object;
                List master = (List) list1.get(2);
                String hostAndPort = new String((byte[]) master.get(0)) + ":" + master.get(1);

                JedisPool jedisPool = nodePollMap.get(hostAndPort);
                slotTree.put((Long) list1.get(0), new PollInfo(jedisPool));
                slotTree.put((Long) list1.get(1), new PollInfo(jedisPool));
            }
        }

        public class PollInfo {
            private JedisPool jedisPool;
            private List<T> dataList = new ArrayList<>(500);

            PollInfo(JedisPool jedisPool) {
                this.jedisPool = jedisPool;
            }

            public JedisPool getJedisPool() {
                return jedisPool;
            }

            public List<T> getDataList() {
                return dataList;
            }
        }
    }

}
